package com.shujia.gma.dwd

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo1PaidOrderDetail {
  /**
    * 支付订单明细表
    *
    * @param args
    */
  def main(args: Array[String]): Unit = {
    val fsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //计划器
      .inStreamingMode()
      .build()

    val fsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    fsEnv.setParallelism(1)
    val fsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(fsEnv, fsSettings)


    fsTableEnv.executeSql(
      """
        |
        |-- -------------------------
        |--   订单详情
        |--   Kafka Source
        |-- -------------------------
        |CREATE TABLE ods_order_detail(
        |  `id` BIGINT,
        |  `order_id` BIGINT,
        |  `sku_id` BIGINT,
        |  `sku_name` STRING,
        |  `img_url` STRING,
        |  `order_price` DECIMAL(10,2),
        |  `sku_num` INT,
        |  `create_time` TIMESTAMP(0)
        |) WITH(
        | 'connector' = 'kafka',
        | 'topic' = 'gma.order_detail',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup1',
        | 'format' = 'canal-json' ,
        | 'scan.startup.mode' = 'earliest-offset'
        |)
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |
        |-- -------------------------
        |--   订单信息
        |--   Kafka Source
        |-- -------------------------
        |CREATE TABLE `ods_order_info` (
        |  `id` BIGINT,
        |  `consignee` STRING,
        |  `consignee_tel` STRING,
        |  `total_amount` DECIMAL(10,2),
        |  `order_status` STRING,
        |  `user_id` BIGINT,
        |  `payment_way` STRING,
        |  `delivery_address` STRING,
        |  `order_comment` STRING,
        |  `out_trade_no` STRING,
        |  `trade_body` STRING,
        |  `create_time` TIMESTAMP(0) ,
        |  `operate_time` TIMESTAMP(0) ,
        |  `expire_time` TIMESTAMP(0) ,
        |  `tracking_no` STRING,
        |  `parent_order_id` BIGINT,
        |  `img_url` STRING,
        |  `province_id` INT
        |) WITH(
        |'connector' = 'kafka',
        | 'topic' = 'gma.order_info',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup3',
        | 'format' = 'canal-json' ,
        | 'scan.startup.mode' = 'earliest-offset'
        |)
        |
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |
        |
        |-- ---------------------------------
        |-- DWD层,支付订单明细表dwd_paid_order_detail
        |-- ---------------------------------
        |CREATE TABLE dwd_paid_order_detail
        |(
        |  detail_id BIGINT,
        |  order_id BIGINT,
        |  user_id BIGINT,
        |  province_id INT,
        |  sku_id BIGINT,
        |  sku_name STRING,
        |  sku_num INT,
        |  order_price DECIMAL(10,0),
        |  create_time TIMESTAMP(0),
        |  pay_time TIMESTAMP(0)
        | ) WITH (
        |    'connector' = 'kafka',
        |    'topic' = 'dwd.paid_order_detail',
        |    'scan.startup.mode' = 'earliest-offset',
        |    'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        |    'format' = 'changelog-json'
        |)
        |
      """.stripMargin)

    fsTableEnv.executeSql(
      """
        |insert into dwd_paid_order_detail
        |SELECT
        |  od.id as detail_id,
        |  oi.id as order_id,
        |  oi.user_id,
        |  oi.province_id,
        |  od.sku_id,
        |  od.sku_name,
        |  od.sku_num,
        |  od.order_price,
        |  oi.create_time,
        |  oi.operate_time as pay_time
        |FROM
        |    (
        |    SELECT *
        |    FROM ods_order_info
        |    WHERE order_status = '2' -- 已支付
        |    ) oi JOIN
        |    (
        |    SELECT *
        |    FROM ods_order_detail
        |    ) od
        |    ON oi.id = od.order_id
        |
      """.stripMargin)

  }

}
